package cn.xinfei.xdecision.pulsar;


import cn.xinfei.xdecision.common.utils.util.JsonUtils;
import cn.xinfei.xdecision.pulsar.config.XDecisionMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClientException;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Slf4j
@Service
public class XPulsarProducer {


    public void send(String topic, XDecisionMessage msg) throws PulsarClientException {
        XPulsarClientFactory.getProducer(topic).send(JsonUtils.toJsonString(msg));
    }

    public CompletableFuture<MessageId> sendAsync(String topic, XDecisionMessage msg) throws PulsarClientException {
        Producer<String> producer = XPulsarClientFactory.getProducer(topic);
        CompletableFuture<MessageId> completableFuture = producer.newMessage()
                .key(msg.getKey())
//                .value(JSONObject.toJSONString(msg))
                .value(JsonUtils.toJsonString(msg))
                .sendAsync();
//        CompletableFuture<MessageId> completableFuture = producer.sendAsync(JSONObject.toJSONString(msg));
        return completableFuture;
    }
}